package com.yanggu.bigdata.realtime.app.dwm

import cn.hutool.json.JSONUtil
import com.alibaba.fastjson.{JSON, JSONException, JSONObject}
import com.yanggu.bigdata.realtime.app.OutputTagObject.errorJsonDataOutputTag
import com.yanggu.bigdata.realtime.app.function.UniqueVisitRichFilterFunction
import com.yanggu.bigdata.realtime.common.GmallConfig.KAFKA_BROKER_LIST
import com.yanggu.bigdata.realtime.utils.KafkaUtil
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * DWM层的独立访问UV
 * 过滤出每日首次访问的PageLog数据
 */
object UniqueVisitApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val jsonDataStream = KafkaUtil.getKafkaDataStream(env, KAFKA_BROKER_LIST, "dwd_page_log", "UniqueVisitApp")
      .process((value: String, ctx: ProcessFunction[String, JSONObject]#Context, out: Collector[JSONObject]) => {
        try {
          val data = JSON.parseObject(value)
          out.collect(data)
        } catch {
          case _: JSONException =>
            ctx.output(errorJsonDataOutputTag, value)
        }
      })
      /*
      {
        "common": {
          "ar": "110000",
          "uid": "38",
          "os": "Android 11.0",
          "ch": "xiaomi",
          "is_new": "1",
          "md": "Xiaomi 9",
          "mid": "mid_14",
          "vc": "v2.1.134",
          "ba": "Xiaomi"
        },
        "page": {
          "page_id": "payment",
          "item": "3,4,7",
          "during_time": 9777,
          "item_type": "sku_ids",
          "last_page_id": "trade"
        },
        "ts": 1608276037000
      }
       */
      //根据mid进行分组
      .keyBy(data => data.getJSONObject("common").getString("mid"))
      //将每次首次登录的数据过滤出来
      .filter(new UniqueVisitRichFilterFunction)
      .map(JSONUtil.toJsonStr(_))

    jsonDataStream.getSideOutput(errorJsonDataOutputTag).print("DirtyData>>>>>>>")
    //控制台打印输出
    jsonDataStream.print("Data>>>>>>")

    //输出到kafka
    jsonDataStream
      .addSink(KafkaUtil.getKafkaSink(KAFKA_BROKER_LIST, "dwm_unique_visit"))

    env.execute("UniqueVisitApp Job")

  }

}
